package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.{HashPartitioner, Partitioner}
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.1 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 * 分区器
 *     在分区的时候使用[groupByKey  groupBy shuffle的时候] , 指定分区规则
 *     判断每个KEY该分到哪个分区 , 返回分区编号
 *    getPartition(key: Any): Int
 */
object C15_ConsumPartitioner {
  def main(args: Array[String]): Unit = {
    /**
     * 加载数据
     */

    val sc = SparkUtil.getSparkContext("自定义分区")
    /**
     * 4,zss3,M,北京,24
      5,zss5,F,南京,27
      6,zss6,M,北京,23
      7,zss7,F,上海,90
     */
    val dataRDD = sc.textFile("data/log/")
    val tpRDD: RDD[(Int, String, String, String, Int)] = dataRDD.map(line => {
      try {
        val arr = line.split(",")
        (arr(0).toInt, arr(1), arr(2), arr(3), arr(4).toInt)
      } catch {
        case e: Exception => null
      }
    })
    // 过滤出不为null的数据
    val filterRDD = tpRDD.filter(_ != null)
    // Array存储所有的城市信息
    // RDD转换成本地集合的方法  collect  收集
    val value: RDD[String] = filterRDD.map(_._4).distinct()
    val citys = value.collect()
   /* for(i <- 0 until citys.length){
      println(s"${citys(i)} 所在的位置是 $i")
    }*/

    // 参数一 分组的字段
    // 参数二  分区器  默认分区
    val grouped = filterRDD.groupBy(tp => tp._4, new HashPartitioner(2))

    // 展示各个分区的数据
    grouped.mapPartitionsWithIndex((p,iter)=>{
      iter.map(tp=>(tp , "分区号是:"+ p))
    }).foreach(println)

    // 打印分区个数
    println("分区个数是: "+ grouped.getNumPartitions)
    println("=================自定义分区器后的数据分配======================")

    /**
     * 参数二  : 自定义分区器
     *  (将城市存储在数组集合中 去重  城市对应的角标 就是分区号 )
     *   每个城市一个分区
     */
    val grouped2 = filterRDD.groupBy(tp => tp._4, new   CityPartitioner(citys))
    grouped2.mapPartitionsWithIndex((p,iter)=>{
      iter.map(tp=>(tp , "分区号是:"+ p))
    }).foreach(println)
  }
}

class  CityPartitioner(citys:Array[String]) extends Partitioner{
  override def numPartitions: Int = citys.length
  override def getPartition(key: Any): Int = {
    val city = key.asInstanceOf[String]
    val index = citys.indexOf(city)
    index
  }
}